package com.zt.bigdata.elasticsearch.rest;

import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.RequestLine;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.*;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;

import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import static java.util.Collections.singletonMap;

/**
 * @ClassName ESRestUtils
 * @Description
 * @Author zhangtonghy
 * @Date 2019-03-11 17:49
 * @Copyright: 版权所有 (C) zt zt.
 * @注意：本内容仅限于zt内部传阅，禁止外泄以及用于其他的商业目的
 **/

public class ESRestUtils {

    private static final String hostname = "";
    private static final String username = "";
    private static final String password = "";
    private static final int port = 9200;

    private static RestHighLevelClient restHighLevelClient = null;


    public static RestHighLevelClient getClient(boolean authorization) {
        if (null == restHighLevelClient) {
            System.out.println("创建客户端");




            RestClientBuilder builder = RestClient.builder(
                    new HttpHost(hostname, port))

                    .setRequestConfigCallback(
                            new RestClientBuilder.RequestConfigCallback() {
                                @Override
                                public RequestConfig.Builder customizeRequestConfig(
                                        RequestConfig.Builder requestConfigBuilder) {
                                    return requestConfigBuilder
                                            .setConnectTimeout(5000)
                                            .setSocketTimeout(60000);
                                }
                            })
                    .setMaxRetryTimeoutMillis(60000);

            if (authorization){
                final CredentialsProvider credentialsProvider =
                        new BasicCredentialsProvider();
                credentialsProvider.setCredentials(AuthScope.ANY,
                        new UsernamePasswordCredentials(username, password));
                builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder
                        .setDefaultCredentialsProvider(credentialsProvider));
            }

            restHighLevelClient = new RestHighLevelClient(builder);
            return restHighLevelClient;
        }
        return restHighLevelClient;
    }

    public static void close() {

        if (null != restHighLevelClient) {
            try {
                System.out.println("关闭客户端");
                restHighLevelClient.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static IndexResponse indexDoc(String index, String type, String id, String jsonContent) throws Exception {

        RestHighLevelClient client = getClient(true);

        IndexRequest indexRequest = new IndexRequest(
                index,
                type, id);

        indexRequest.source(jsonContent, XContentType.JSON);
//        indexRequest.routing(index);

        return client.index(indexRequest, RequestOptions.DEFAULT);

    }


    public static IndexResponse indexDoc(String index, String type, String id, Map<String, Object> jsonContent) throws Exception {

        IndexRequest indexRequest = new IndexRequest(index, type, id)
                .source(jsonContent);
//        indexRequest.routing(index);
        indexRequest.timeout(TimeValue.timeValueSeconds(1));

        return getClient(true).index(indexRequest, RequestOptions.DEFAULT);
    }

    public static IndexResponse indexDoc(String index, String type, String id, XContentBuilder xContentBuilder) throws Exception {

        IndexRequest indexRequest = new IndexRequest(index, type, id)
                .source(xContentBuilder);
        indexRequest.timeout("1s");
//        indexRequest.routing(index);
        indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
        return getClient(true).index(indexRequest, RequestOptions.DEFAULT);
    }

    public static void deleteDoc(String index, String type, String id) throws Exception {

        DeleteRequest deleteRequest = new DeleteRequest(index, type, id);
        deleteRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
        DeleteResponse deleteResponse = getClient(true).delete(
                deleteRequest, RequestOptions.DEFAULT);
        System.out.println(deleteResponse.status().getStatus());

        ReplicationResponse.ShardInfo shardInfo = deleteResponse.getShardInfo();
        if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
            System.out.println("分片个数不等于成功分片个数");
        }
        if (shardInfo.getFailed() > 0) {
            for (ReplicationResponse.ShardInfo.Failure failure :
                    shardInfo.getFailures()) {
                String reason = failure.reason();
                System.out.println("失败原因：" + reason);
            }
        }

    }


    public static void asyncIndex(String index, String type, String id, XContentBuilder xContentBuilder) throws Exception {

        ActionListener listener = new ActionListener<IndexResponse>() {
            @Override
            public void onResponse(IndexResponse indexResponse) {
                System.out.println("异步返回: " + indexResponse.getIndex() + " " + indexResponse.getType() + " " + indexResponse.getId());
            }

            @Override
            public void onFailure(Exception e) {
                e.printStackTrace();
            }
        };

        IndexRequest indexRequest = new IndexRequest(index, type, id)
                .source(xContentBuilder);
        indexRequest.timeout("1s");
//        indexRequest.routing(index);
        indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
        getClient(true).indexAsync(indexRequest, RequestOptions.DEFAULT, listener);

    }

    public static GetResponse get(String index, String type, String id) throws IOException {
        GetRequest request = new GetRequest(index, type, id);
        GetResponse getResponse = getClient(true).get(request, RequestOptions.DEFAULT);
        return getResponse;
    }

    public static UpdateResponse updateByScript(String index, String type, String id) throws IOException {
        UpdateRequest updateRequest = new UpdateRequest(index, type, id);

        Map<String, Object> parameters = singletonMap("count", 4);

        Script inline = new Script(ScriptType.INLINE, "painless",
                "ctx._source.age += params.count", parameters);

        updateRequest.script(inline);

        UpdateResponse updateResponse = getClient(true).update(updateRequest, RequestOptions.DEFAULT);
        return updateResponse;
    }

    public static GetResponse get(String index, String type, String id, boolean fetchSource, String[] includes, String[] excludes) throws IOException {
        GetRequest getRequest = new GetRequest(index, type, id);
        getRequest.routing("routing");
        FetchSourceContext fetchSourceContext =
                new FetchSourceContext(fetchSource, includes, excludes);
        getRequest.fetchSourceContext(fetchSourceContext);
        GetResponse getResponse = getClient(true).get(getRequest, RequestOptions.DEFAULT);
        return getResponse;
    }

    public static boolean exists(String index, String type, String id) throws IOException {
        GetRequest getRequest = new GetRequest(index, type, id);
        FetchSourceContext fetchSourceContext =
                new FetchSourceContext(false);
        getRequest.fetchSourceContext(fetchSourceContext);
        getRequest.storedFields("_none_");
        boolean exists = getClient(true).exists(getRequest, RequestOptions.DEFAULT);
        return exists;
    }

    public static Response request(String httpMethod, String uri) throws Exception {

        Response response = getClient(true)
                .getLowLevelClient()
                .performRequest(new Request(httpMethod, uri));

        return response;
    }


    public static void main(String[] args) throws Exception {
//        String jsonString = "{" +
//                "\"user\":\"kimchy\"," +
//                "\"postDate\":\"2013-01-30\"," +
//                "\"message\":\"trying out Elasticsearch\"" +
//                "}";
//        IndexResponse indexResponse = indexDoc("posts", "doc", "1", jsonString);
//        System.out.println(indexResponse.getIndex() + " " + indexResponse.getType() + " " + indexResponse.getId());
//
//
//        Map<String, Object> jsonMap = new HashMap(4);
//        jsonMap.put("user", "kimchy");
//        jsonMap.put("age", 20);
//        jsonMap.put("postDate", new Date());
//        jsonMap.put("message", "trying out Elasticsearch");
//        indexResponse = indexDoc("posts", "doc", "1", jsonMap);
//        System.out.println(indexResponse.getIndex() + " " + indexResponse.getType() + " " + indexResponse.getId());
//
//        XContentBuilder builder = XContentFactory.jsonBuilder();
//        builder.startObject();
//        {
//            builder.field("user", "kimchy");
//            builder.field("age", 20);
//            builder.timeField("postDate", new Date());
//            builder.field("message", "trying out Elasticsearch");
//        }
//        builder.endObject();
//        indexResponse = indexDoc("posts", "doc", "1", builder);
//        System.out.println(indexResponse.getIndex() + " " + indexResponse.getType() + " " + indexResponse.getId());
//
////        asyncIndex("posts", "doc", "1", builder);
//
//        GetResponse getResponse = get("posts", "doc", indexResponse.getId());
//
//        if (getResponse.isExists()) {
//            String index = getResponse.getIndex();
//            String type = getResponse.getType();
//            String id = getResponse.getId();
//            long version = getResponse.getVersion();
//            String sourceAsString = getResponse.getSourceAsString();
//            Map<String, Object> sourceAsMap = getResponse.getSourceAsMap();
//            byte[] sourceAsBytes = getResponse.getSourceAsBytes();
//        } else {
//            System.out.println("null");
//        }
//
//        updateByScript("posts", "doc", "1");
//        System.out.println(get("posts", "doc", "1").getSourceAsMap().get("age"));
//
//        deleteDoc("posts", "doc", "1");
//
//
//        Response response = request("GET", "/");
//        RequestLine requestLine = response.getRequestLine();
//        HttpHost host = response.getHost();
//        int statusCode = response.getStatusLine().getStatusCode();
//        Header[] headers = response.getHeaders();
//        String responseBody = EntityUtils.toString(response.getEntity());
//        System.out.println(host + " " + statusCode + " \n" + responseBody);
//
//        close();


        SearchRequest searchRequest = new SearchRequest("time_series");
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(QueryBuilders.matchAllQuery());
        searchSourceBuilder.size(1000);
        searchRequest.source(searchSourceBuilder);

        try {
            SearchResponse searchResponse =  getClient(true).search(searchRequest, RequestOptions.DEFAULT);
            System.out.println(searchResponse.getHits().getHits().length);
        } catch (IOException e) {
            e.printStackTrace();

        }
    }
}
